﻿#region Copyright (c) 2013-04, Olaf Kober <amarok.projects@gmail.com>
//================================================================================
//	Permission is hereby granted, free of charge, to any person obtaining a copy
//	of this software and associated documentation files (the "Software"), to deal
//	in the Software without restriction, including without limitation the rights
//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
//	copies of the Software, and to permit persons to whom the Software is
//	furnished to do so, subject to the following conditions:
//
//	The above copyright notice and this permission notice shall be included in
//	all copies or substantial portions of the Software.
//
//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
//	THE SOFTWARE.
//================================================================================
#endregion

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;


namespace Amarok.Agents
{
	/// <summary>
	/// This type implements a dispatcher, which is an object that is capable to forward (dispatch) messages to 
	/// appropriate message handler. In general this is done by invoking the message handler method and supplying 
	/// the message as an argument to the invoked method. In addition, a dispatcher provides rich queuing and 
	/// scheduling mechanism that decouple caller from callee. Messages posted to the dispatcher are first enqueued 
	/// into an internal queue and then later forwarded to their respective handlers.
	/// </summary>
	public sealed class DefaultDispatcher : CompletableObject,
		IDispatcher
	{
		// data
		private readonly CancellationTokenSource mCancellationTokenSource =
			new CancellationTokenSource();
		private readonly Object mSyncIsInitialized =
			new Object();

		private readonly ConcurrentExclusiveSchedulerPair mSchedulerPair;
		private readonly HandlerEntry[] mHandlerEntries;
		private readonly ActionBlock<None> mInitializeActionBlock;
		private readonly Func<Message, Boolean> mFilterFunc;

		// state
		private volatile Boolean mIsInitialized;


		#region ++ ICompletable Interface ++

		/// <summary>
		/// Gets a Task instance that represents the completion state of the object. This task instance is 
		/// transitioned into the final RunToCompletion state when the object has finally completed.
		/// </summary>
		public new Task Completion
		{
			get
			{
				return base.Completion;
			}
		}


		/// <summary>
		/// Initiates the asynchronous completion of the object. After calling this method the object will complete 
		/// and its <see cref="Completion"/> task will enter a final state after it has processed or discarded all 
		/// previously available messages.
		/// </summary>
		/// 
		/// <param name="discardAvailableMessages">
		/// True to discard all already available messages. False to process all already available messages to 
		/// completion. Defaults to false.</param>
		public void Complete(Boolean discardAvailableMessages = false)
		{
			base.CompleteObject(discardAvailableMessages);
		}

		#endregion

		#region ++ IDispatcher Interface ++

		/// <summary>
		/// Posts the message to the dispatcher.
		/// 
		/// The method stores the supplied message in the dispatcher's input message queues of interested consumers and 
		/// returns immediately, meaning it only schedules the message for further processing, but the actual processing 
		/// is done later in other threads.
		/// 
		/// Messages that the dispatcher is unable to handle, i.e. because no corresponding message handler exists, 
		/// will be dropped silently.
		/// 
		/// If there is no interested consumer at the time of posting, then the message will be dropped silently.
		/// 
		/// Messages that are posted after completing the dispatcher will be dropped silently.
		/// </summary>
		/// 
		/// <param name="message">
		/// The message that should be posted.</param>
		/// 
		/// <returns>
		/// A boolean value indicating whether the message being posted has been scheduled for further processing, or 
		/// False, if the message has been filter, the dispatcher is completing or already completed, or no matching 
		/// message handler exists.
		/// </returns>
		/// 
		/// <exception cref="ArgumentNullException">
		/// A null reference was passed to a method that did not accept it as a valid argument.</exception>
		public Boolean Post(Message message)
		{
			Verify.NotNull(message, "message");

			// stop processing new messages
			if (base.IsCompleting)
				return false;

			// filter message
			if (mFilterFunc != null)
			{
				if (mFilterFunc(message) == false)
					return false;
			}

			// process message
			Int32 count = 0;
			for (Int32 i = 0; i < mHandlerEntries.Length; i++)
			{
				var entry = mHandlerEntries[i];

				if (entry.Match(message) == false)
					continue;

				// one-time initialization
				if (mIsInitialized == false)
					_Initialize();

				// schedule processing
				entry.Post(message);
				count++;
			}

			return count > 0;
		}

		#endregion

		#region ++ Public Interface ++

		/// <summary>
		/// Initializes a new instance.
		/// </summary>
		/// 
		/// <param name="target">
		/// The object with message handler for which the dispatcher should set up message dispatching.</param>
		/// <param name="initializeAction">
		/// An optional initialization callback that is called when the first message arrives. The callback is 
		/// called before the corresponding message handler is invoked. Optional, defaults to null.</param>
		/// <param name="filterFunc">
		/// An optional filter callback that is called for each message. Message for which the filter callback
		/// returns false will be discarded. Optional, defaults to null.</param>
		/// 
		/// <exception cref="ArgumentNullException">
		/// A null reference was passed to a method that did not accept it as a valid argument.</exception>
		public DefaultDispatcher(
			Object target,
			Action initializeAction = null,
			Func<Message, Boolean> filterFunc = null)
		{
			Verify.NotNull(target, "target");

			mSchedulerPair = new ConcurrentExclusiveSchedulerPair(
				TaskScheduler.Default,
				AgentGlobals.MaxConcurrencyLevel,
				AgentGlobals.MaxMessagesPerTask);

			var handlerInfos = Inspector.Inspect(
				target.GetType());

			mHandlerEntries = _BuildHandlerEntries(
				handlerInfos,
				target);

			mInitializeActionBlock = _BuildInitializeActionBlock(
				initializeAction);

			mFilterFunc = filterFunc;
		}

		#endregion

		#region Implementation

		/// <summary>
		/// This overrideable method supports the agent infrastructure and is not intended to be used directly.
		/// </summary>
		protected override Task OnComplete(Boolean discardAvailableMessages)
		{
			var taskSource = new TaskCompletionSource();

			// cancel handler; discard pending messages
			if (discardAvailableMessages)
				mCancellationTokenSource.Cancel();

			// complete handler
			var completionTasks = _CompleteHandler();
			TaskHelper.WhenAll(completionTasks)
				.ContinueWith(t1 =>
					{
						// complete scheduler
						mSchedulerPair.Complete();
						mSchedulerPair.Completion.ContinueWith(t2 =>
							{
								// done
								taskSource.SignalCompletion();
							},
							TaskContinuationOptions.ExecuteSynchronously);
					},
					TaskContinuationOptions.ExecuteSynchronously);

			return taskSource.Task;
		}

		/// <summary>
		/// This overrideable method supports the agent infrastructure and is not intended to be used directly.
		/// </summary>
		protected override void OnCompleted()
		{
			// done
			Helper.Dispose(mCancellationTokenSource);
		}


		private Task[] _CompleteHandler()
		{
			var completionTasks = new Task[mHandlerEntries.Length + 1];

			mInitializeActionBlock.Complete();
			completionTasks[0] = mInitializeActionBlock.Completion;

			for (Int32 i = 0; i < mHandlerEntries.Length; i++)
				completionTasks[1 + i] = mHandlerEntries[i].Complete();

			return completionTasks;
		}

		private void _Initialize()
		{
			// already initialized?
			if (mIsInitialized)
				return;

			lock (mSyncIsInitialized)
			{
				// already initialized?
				if (mIsInitialized)
					return;

				// set flag
				mIsInitialized = true;
			}

			// schedule initialization action (guaranteed to be called once)
			mInitializeActionBlock.Post(None.Instance);
		}

		private ActionBlock<None> _BuildInitializeActionBlock(Action initializeAction)
		{
			var options = new ExecutionDataflowBlockOptions()
			{
				BoundedCapacity = 1,
				MaxMessagesPerTask = 1,
				MaxDegreeOfParallelism = 1,
				NameFormat = "Dispatcher-InitializeActionBlock: Id: {1}",
				TaskScheduler = mSchedulerPair.ExclusiveScheduler,
			};

			return new ActionBlock<None>(
				unit =>
				{
					if (initializeAction != null)
						initializeAction();
				},
				options);
		}

		private HandlerEntry[] _BuildHandlerEntries(HandlerInfo[] handlerInfos, Object target)
		{
			var entries = new HandlerEntry[handlerInfos.Length];

			for (Int32 i = 0; i < handlerInfos.Length; i++)
			{
				var actionBlock = _BuildActionBlock(
					handlerInfos[i],
					target);

				entries[i] = new HandlerEntry(
					handlerInfos[i].MessageType,
					actionBlock);
			}

			return entries;
		}

		private ActionBlock<Message> _BuildActionBlock(HandlerInfo handlerInfo, Object target)
		{
			var options = new ExecutionDataflowBlockOptions()
				{
					NameFormat = "Dispatcher-ActionBlock: Id: {1}",
					BoundedCapacity = AgentGlobals.BoundedCapacity,
					CancellationToken = mCancellationTokenSource.Token,
					MaxMessagesPerTask = _GetMaxMessagesPerTask(handlerInfo),
					MaxDegreeOfParallelism = _GetMaxDegreeOfParallelism(handlerInfo),
					TaskScheduler = _GetScheduler(handlerInfo),
				};

			var handler = handlerInfo.CreateHandler(target);
			handler.Initialize(target as IAgent);
			var handlerDelegate = (Action<Message>)handler.Invoke;
			return new ActionBlock<Message>(handlerDelegate, options);
		}

		private Int32 _GetMaxDegreeOfParallelism(HandlerInfo handlerInfo)
		{
			if (handlerInfo.ExecutionMode == ExecutionMode.Exclusive)
				return 1;

			return handlerInfo.ParallelismLevel;
		}

		private Int32 _GetMaxMessagesPerTask(HandlerInfo handlerInfo)
		{
			if (handlerInfo.ExecutionMode == ExecutionMode.Exclusive)
				return 1;

			return AgentGlobals.MaxMessagesPerTask;
		}

		private TaskScheduler _GetScheduler(HandlerInfo handlerInfo)
		{
			switch (handlerInfo.ExecutionMode)
			{
			case ExecutionMode.Independent:
				return TaskScheduler.Default;
			case ExecutionMode.Concurrent:
				return mSchedulerPair.ConcurrentScheduler;
			case ExecutionMode.Exclusive:
				return mSchedulerPair.ExclusiveScheduler;
			default:
				return TaskScheduler.Default;
			}
		}

		#endregion

		#region Implementation (Nested Types)

		private struct HandlerEntry
		{
			private readonly Type mType;
			private readonly ActionBlock<Message> mActionBlock;


			public HandlerEntry(Type type, ActionBlock<Message> actionBlock)
			{
				mType = type;
				mActionBlock = actionBlock;
			}

			public Boolean Match(Message message)
			{
				return mType.IsInstanceOfType(message);
			}

			public void Post(Message message)
			{
				mActionBlock.Post(message);
			}

			public Task Complete()
			{
				mActionBlock.Complete();
				return mActionBlock.Completion;
			}
		}

		#endregion

	}
}
